package bigdata.backend.test;

import bigdata.backend.monitor.utils.KafkaProgressUtils;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
public class TestKafkaProgressUtils {

    @Test
    public void test01() {
        String bootstrapServers = "localhost:9092";
        String consumerGroupId = "binlog-monitor-group";
        String topicName = "binlog-monitor";

        // 每次调用都会创建新的连接，查询后立即关闭
        KafkaProgressUtils.TopicProgressInfo progress = KafkaProgressUtils.getTopicProgress(bootstrapServers, consumerGroupId, topicName);

        if (progress.getError() != null) {
            System.err.println("查询失败: " + progress.getError());
        } else {
            System.out.println("\n--- 查询结果 ---");
            System.out.println("Topic: " + progress.getTopicName());
            System.out.println("总消费: " + progress.getTotalConsumed());
            System.out.println("总消息: " + progress.getTotalMessages());
            System.out.println("总延迟: " + progress.getTotalLag());

            System.out.println("分区详情:");
            for (KafkaProgressUtils.TopicProgressInfo.PartitionProgress p : progress.getPartitionProgressList()) {
                System.out.println("  Partition " + p.getPartition() +
                        ": consumed=" + p.getConsumedMessages() +
                        ", total=" + p.getTotalMessages() +
                        ", lag=" + p.getLag());
            }
        }
    }
}
